Repository navigation
Conversation
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
run buildall |
|
/review |
888fa07 to
a3c9316
Compare
|
run buildall |
|
/review |
TPC-H: Total hot run time: 28452 ms |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Static review of PR 69850 converged after two rounds. The wrapper row-count delegation correctly fixes reset and snapshot stream estimates for regular OlapTableWrapper paths, and the changelog UNKNOWN_ROW_COUNT contract remains intact for ordinary whole-scan callers. One P2 remains: delegating getPartitionNum also activates StatsCalculator's selected-partition fallback for OlapChangelogTableWrapper, so a direct @incr or incremental stream scan over a strict partition subset estimates one row per selected partition even when the offset range is empty or tiny; the added tests do not cover that multi-partition path. No existing inline findings were present. Relevant checkpoints: the change is focused and the main reset/snapshot behavior is covered by unit and regression tests; no new concurrency, lifecycle, configuration, protocol, persistence, or observability path was introduced; all wrapper construction and cloud/shared-nothing branches were traced; tests/builds were not executed because this review contract is static-only. No additional user review focus was supplied.
Review converged after 2 round(s).
| } | ||
|
|
||
| @Override | ||
| public int getPartitionNum() { |
There was a problem hiding this comment.
[P2] Preserve the changelog wrapper's unknown cardinality for partition-pruned reads. With PARTITION (p1, p2) on a direct @incr scan (or an incremental stream plan with a strict subset), this new getPartitionNum() makes StatsCalculator enter its selected-partition fallback. Both getRowCountForPartitionIndex calls return UNKNOWN_ROW_COUNT, so the fallback fabricates one row per selected partition; a range can contain no change rows, and the estimate now grows with the number of selected partitions. Before this override the wrapper partition count was zero, the branch was skipped, and the estimate stayed at the existing one-row unknown fallback. Keep changelog scans on the unknown path (or bypass this partition fallback) and add a multi-partition incremental EXPLAIN regression.
There was a problem hiding this comment.
Keeping the delegation and the changelog estimate as is. OlapChangelogTableWrapper still answers UNKNOWN_ROW_COUNT for both row-count methods, so the change here is only which of the statistics calculator's two generic lower bounds applies to a partition-pruned changelog scan: one row for the relation before, one row per selected partition after. Both numbers describe a quantity that is unknown by design, both stay orders of magnitude below broadcast_row_count_limit, and neither moves the join or broadcast choice -- the estimate does not grow with the size of the change range, which is what would matter. Suppressing the per-partition fallback for this wrapper would put a catalog-type check into the generic statistics path, so I'd rather track "unknown relations should not be broadcast build sides" separately.
I did not add a multi-partition incremental EXPLAIN case: the value it would pin down is exactly the fabricated estimate (1 vs number of selected partitions), and a regression that asserts an estimate the change does not intend to make more accurate would be a maintenance liability. If you'd rather have the old one-row behaviour preserved bit-for-bit, say so and I'll bypass the fallback for the changelog wrapper instead.
TPC-DS: Total hot run time: 150532 ms |
ClickBench: Total hot run time: 24.35 s |
TPC-H: Total hot run time: 28702 ms |
TPC-DS: Total hot run time: 151042 ms |
ClickBench: Total hot run time: 24.36 s |
There was a problem hiding this comment.
Static review of PR 69850 converged after two rounds. The wrapper row-count delegation and changelog UNKNOWN_ROW_COUNT overrides were traced through StatsCalculator, stream binding, snapshot/reset planning, and all row-count consumers. One new P2 is reported in the regression fixture: it creates the stream immediately after the initial insert without synchronizing partition visibility, so local stream creation can capture TSO -1 and snapshot normalization can produce no PhysicalOlapScan or skip the wrapper path. Add the established sync/visibility waits before CREATE STREAM and before the post-create explain checks. Existing inline P2 discussion 4414336957 remains confirmed: delegated getPartitionNum() activates the selected-partition fallback for changelog UNKNOWN_ROW_COUNT; it is not reposted. All four changed files and the full regression DSL/explain path were reviewed. The user focus file specifies no additional focus. Per the static-only contract, no builds, tests, or runtime EXPLAIN were executed.
Review converged after 2 round(s).
| // Five rows exist when the stream is created; the last two are written afterwards, so the | ||
| // snapshot image holds five rows while the table holds seven. | ||
| sql "INSERT INTO stats_stream_t VALUES (1,10),(2,20),(3,30),(4,40),(5,50)" | ||
| sql "CREATE STREAM stats_stream ON TABLE stats_stream_t PROPERTIES ('show_initial_rows'='false')" |
There was a problem hiding this comment.
[P2] Synchronize the initial insert before creating the stream. The test inserts five rows and immediately executes CREATE STREAM here, but a successful INSERT can remain committed while the local partition TSO is still -1 until publish completes. show_initial_rows=false records that TSO; snapshot filtering accepts only positive offsets, so stats_stream@snapshot() can normalize to an empty relation with no PhysicalOlapScan, causing the helper to fail or skip the wrapper path. Add the established sync/visibility wait before CREATE STREAM (and synchronize the post-create insert before explaining) so this regression deterministically captures the five-row snapshot.
There was a problem hiding this comment.
Fixed in 0276aff: sql "sync" now follows both inserts, so the rows are published before CREATE STREAM records the stream's offset and before the post-create inserts are explained.
FE Regression Coverage ReportIncrement line coverage |
2 similar comments
FE Regression Coverage ReportIncrement line coverage |
FE Regression Coverage ReportIncrement line coverage |
0276aff to
499a3aa
Compare
499a3aa to
19a6559
Compare
|
run buildall |
|
/review |
There was a problem hiding this comment.
Static review of PR 69850 converged after 2 rounds. The new stream row-count estimates are wired through wrapper construction, StatsCalculator, changelog TTL/TSO handling, snapshot/reset/rebuild branches, local and cloud offset publication, persistence, and partition pruning. One P1 remains: the planner can capture an old stream offset range and later read a newer partition baseline without an atomic stream snapshot, and replayed state maps bypass the intended concurrent-map initialization; this can under-estimate a nonempty stream read and recreate the broadcast-build memory risk. Fix by capturing the range and baseline together under the stream lock or an immutable/versioned snapshot, and normalize replayed state. Existing inline P2 4414336957 remains confirmed for the changelog partition fallback and is intentionally not duplicated. The regression-fixture P2 was fixed in the current history. No additional user focus was supplied. Static-only validation applies: no builds, tests, or runtime EXPLAINs were executed.
Review converged after 2 round(s).
| partitionOffset.put(entry.getKey(), entry.getValue()); | ||
| partitionConsumptionTime.put(entry.getKey(), ts); | ||
| } | ||
| unprotectedRecordPartitionStreamStates(next.keySet()); |
There was a problem hiding this comment.
[P1] Keep the stream range and row-count baseline from one atomic snapshot. OlapTableStreamWrapper builds outputUpdateMap from stream.getStreamUpdate() without the stream lock, then estimatePartitionChangelogRead() reads getPartitionStreamState() later. A concurrent unprotectedUpdateStreamUpdate() can advance the offset and replace the baseline at this line, so the planner pairs the old (start,end] range with the post-advance counter; changelogDelta() can then return zero or a stale low count for a nonempty range while FE row-count reports lag, recreating the one-row/broadcast-build risk this change is meant to fix. Planning locks the base OlapTable, not the OlapTableStream, and after restart the persisted non-null Map bypasses the ConcurrentHashMap initializer. Capture offset and state under one stream read lock or versioned immutable snapshot and normalize deserialized state to a concurrent map.
There was a problem hiding this comment.
Fixed in a9e7634, both halves of the finding:
-
The range and the counters now come from one read.
OlapTableStreamWrappersnapshots the partitions'PartitionStreamStateat the same time it buildsoutputUpdateMap(constructor, andinstallCloudReadStatesfor cloud read states), with the stream read lock held across both, and the estimate reads that snapshot instead of asking the stream later. An offset advance can no longer pair the range of an old offset with the counters of a newer one. -
The map read back from an image is made concurrent.
OlapTableStream.gsonPostProcess()replaces whatever map the deserialization built with aConcurrentHashMap, andpartitionStates()also converts on first use, so the planner's lock-free reads and the offset advance's replacement do not race on a plain map.
There was a problem hiding this comment.
Static review of PR 69850 converged after two rounds. The wrapper row-count delegation, stream-lock snapshot, deserialized-map normalization, local/cloud publication paths, and reset/snapshot/incremental Normalize branches were traced across all 13 changed files. The prior P1 thread 4423124345 is fixed at this head; existing P2 thread 4414336957 remains confirmed for the partition-pruned unknown-cardinality fallback and is not reposted. The regression-fixture P2 was fixed in the current history.
One verified P1 remains: when a positive changelog interval is known but tablet and table row counters are all temporarily zero, the new fallback still estimates one row, so a large incremental stream can become a broadcast build side and exhaust memory. The accepted inline comment identifies the exact call chain and fix direction. No additional user review focus was supplied. Added unit and regression tests were inspected but not executed, consistent with the static-only contract; runtime counter-publication timing remains the only coverage limit.
Relevant checkpoints: the change is focused and addresses the original one-row estimate goal but leaves the all-counters-zero nonempty-range gap; concurrency and lifecycle paths were rechecked with no additional issue; no configuration, protocol, or thrift/protobuf change is involved; persistence and local/cloud transaction paths were traced; the remaining issue is a planner data-correctness risk rather than a runtime data-loss path.
Review converged after 2 rounds.
Review converged after 2 round(s).
| delta = tableRowCount; | ||
| } | ||
| } | ||
| return Math.max(1, delta); |
There was a problem hiding this comment.
[P1] Keep a known nonempty changelog range from collapsing to one row when counters are unavailable. estimatePartitionChangelogRead() can pass a positive (startTso,endTso] while both tablet counters and getBaseTableRowCount() are still zero (for example, a large load immediately after the stream offset, before backend reports or analysis stats arrive). changelogDelta() then returns UNKNOWN, the fallback leaves delta at zero, and Math.max(1, delta) reports one row even though the changelog scan reads the entire interval; the optimizer can choose that large stream as a broadcast build and exhaust memory. Preserve an unknown/conservative estimate or defer the decision when the interval is known but every counter is unavailable, and add an integration case for planning before tablet statistics are published.
There was a problem hiding this comment.
Fixed in 6d2b951. When the TSOs say the range holds changes and every counter is unavailable (the changelog count, the partition's base count and the table record are all zero), estimateChangelogRead no longer falls through to Math.max(1, ...): it reports UNKNOWN_NONEMPTY_ROW_COUNT (1L << 30), a relation deliberately too large to be broadcast, because the only thing known there is that the range is not empty. An empty range still answers with one row, which the TSO comparison decides without the counters. PartitionStreamStateTest pins both cases (testAKnownNonemptyRangeWithoutCountersIsNotEstimatedAtOneRow), including the unreported-counts case that used to expect one row.
The remaining P2 thread on the partition-pruned fallback is unchanged by design: with the estimates in place the changelog wrapper no longer reports UNKNOWN_ROW_COUNT for a stream read, so the fabrication it warns about only applies to a wrapper without a recorded estimate.
a9e7634 to
6d2b951
Compare
|
run buildall |
|
/review |
### What problem does this PR solve? Issue Number: None Related PR: apache#69850 Problem Summary: A table stream estimates the change rows a read returns from the row count of the partition's changelog index: a read of the range between the stream offset and the end of the read returns at most the change rows the changelog still holds. The count the frontend holds comes from the tablet reports, which a backend sends every tablet_stat_update_interval_second (60s by default), so a read which follows a load within that interval sees the count from before the load. The estimate falls back to the row count of the table, which is safe but loose: it answers with the whole table where the answer is a few change rows. The frontend cannot derive the change rows from the rows it loaded into the base index: an update writes change rows without adding a base row, and a delete adds none to the base but at least one to the changelog. The backend knows them, so it now reports them. - The publish task reports the delta rows of the changelog tablets as well, not only of the base tablets: the report already visits the rowsets of every tablet a transaction wrote, and a changelog tablet is one of them. - AnalysisManager.replayUpdateRowsRecord routes those tablets to the table record by the index of each tablet (from the tablet inverted index, which the followers have as well, so a replay reproduces the routing without a new journal entry), and TableStatsMeta accumulates them per index (Gson field "iur"). The base index rows are counted as before: the change rows are neither added to them nor mistaken for them. - TableStatsMeta.getRowCountWithDeltaRows returns the accumulated rows for a changelog index: an analysis never collects its row count, so the rows of the loads are the count there. A stream read consumes this through TableStatsMeta.getRowCountWithDeltaRows, which is where the estimate of PR apache#69850 reads its counters from; until that lands, the recorded rows are visible in the table statistics only. ### Release note None ### Check List (For Author) - Test: Unit test - Unit test: TableStatsMetaTest, a new case (the rows a load writes to the changelog index are accumulated per index and reported by getRowCountWithDeltaRows, while the base index rows are not affected). TableStatsMetaTest 16/16 and AnalysisManagerTest 27/27 pass with the change. - No regression exercises the reporting window: it needs a load followed by a read within the tablet reporting interval, which the suites cannot time reliably. Every load now reports the change rows as well, so the existing suites which read the row counts of a loaded table cover the base index side of the change. - Behavior changed: No user visible change; a load additionally reports the rows it wrote to the changelog index. - Does this need documentation: No
|
run buildall |
|
/review |
There was a problem hiding this comment.
Static review of PR 69850 found six verified P1 cardinality underestimation paths that can make a large stream read a broadcast build: unknown-counter snapshots, first loads into empty-at-create partitions, partial TTL reclamation, shared-nothing post-offset sampling, cloud idempotent retry re-sampling, and the cloud read-state RPC-to-install window. The wrapper/statistics delegation, selected-partition behavior, TSO conversion, local state locking, Gson old-image normalization, stale cleanup, and cloud visible-version paths were otherwise traced. Existing P1 threads 4423124345 and 4424439892 were independently revalidated as fixed at this head; their IDs are not repeated. The existing P2 threads were treated as known context, with the regression sync issue fixed in the current head. No additional user review focus was provided. Static-only review: no builds, tests, benchmarks, or runtime validation were executed. The added tests cover serial arithmetic/planner cases, but do not cover the accepted concurrent producer/report, cloud retry, RPC/install, unknown-counter snapshot, empty-at-create first-load, or TTL purge/growth scenarios. Critical checkpoints: the task goal is only partially achieved because the new estimates remain unsafe in those paths; concurrency and lifecycle boundaries are the source of all six findings; no protocol/configuration/thrift changes were introduced; test coverage is incomplete for the reported races. Review converged after round 2.
Review converged after 2 round(s).
| if (estimate <= 0) { | ||
| estimate = tableRowCount; | ||
| } | ||
| return Math.max(1, estimate); |
There was a problem hiding this comment.
[P1] Keep an unknown snapshot from becoming a one-row build side. For a partition with a positive stream offset and real rows, initializeLocalOffsets() and getBaseTableRowCount() can both see unreported counters as 0, so estimateSnapshotRead(true, 0, 0, UNKNOWN, 0) reaches this Math.max(1, estimate) and returns 1. The snapshot wrapper then overrides the delegated table count with that one-row estimate, allowing a large image to be chosen as a broadcast build. Preserve a conservative unknown/nonempty estimate when all counters are unavailable and add a pre-report snapshot-planning case.
| if (delta <= 0) { | ||
| delta = tableRowCount; | ||
| } | ||
| if (delta <= 0 && startTso >= 0 && endTso > startTso) { |
There was a problem hiding this comment.
[P1] Do not treat an empty-at-create partition's first load as a one-row range. Its stored offset is -1, which toExclusiveBound converts to an unset start; once the partition receives a large first load, filterIncrementalPartitionIds() still plans the changelog scan from the beginning. With counters and table stats still 0, this startTso >= 0 guard is false and Math.max(1, delta) reports one row, so the full first-load range can become a broadcast build. Treat an unknown start with a known nonempty/end range conservatively and add this case to the regression coverage.
| if (endTso <= startTso) { | ||
| return 0; | ||
| } | ||
| if (storedChangelogRowCount >= 0 && currentChangelogRowCount > storedChangelogRowCount) { |
There was a problem hiding this comment.
[P1] Keep the delta conservative across partial TTL reclamation. The baseline is recorded before old changelog rowsets are removed, while applyChangelogTtl() clips the scan start to the first retained TSO. If the baseline is 100, TTL purges 90 old rows, and 110 new rows are appended, the current count is 120 and this branch returns 20 even though the clipped range can contain 110 rows. That under-estimate can make a large range a broadcast build. Detect a reclaimed/TTL-clipped baseline or use a retained-count bound, and add a purge-then-growth test.
| partitionOffset.put(entry.getKey(), entry.getValue()); | ||
| partitionConsumptionTime.put(entry.getKey(), ts); | ||
| } | ||
| unprotectedRecordPartitionStreamStates(next.keySet()); |
There was a problem hiding this comment.
[P1] Do not sample counters after advancing the stream offset without a source-TSO fence. The consumer commit locks the sink and stream, but not the stream's base table; a producer can commit/report rows with TSO greater than the captured next between the offset write and this call. The stored baseline then already includes rows from the next (startTso,endTso] range, so current - stored can be zero/one and select a large changelog as a broadcast build. Capture counters at the offset boundary or version/filter them by source TSO, and add a producer/consumer race test.
| try { | ||
| txnState = commitTxn(builder, tableList, transactionId, is2PC, tabletCommitInfos, tabletIds); | ||
| txnOperated = true; | ||
| recordStreamPartitionRowCounts(builder.getTableStreamUpdatesList()); |
There was a problem hiding this comment.
[P1] Skip baseline recording when a cloud commit is only an idempotent retry. After a persisted transaction is already visible, MetaService returns before applying the table-stream update, but commitTxn() accepts that response and this line records the builder's partitions anyway. Rows/reporting that arrive between the original commit and the retry are then stored as the baseline for the unchanged offset, so the next current - stored estimate omits them and can choose a broadcast build. Record only on the transition or make the update keyed/idempotent, and add a duplicate-commit test.
| outputUpdateMap.put(partitionId, Pair.of(startTso, state.getEndTso())); | ||
| } | ||
| }); | ||
| partitionStreamStates = snapshotPartitionStreamStates(outputUpdateMap.keySet()); |
There was a problem hiding this comment.
[P1] Fence the cloud read-state snapshot with the local baseline. CloudTableStreamReadStateHook fetches the authoritative range from MetaService before calling this method; a concurrent consumer can advance the stream and replace partitionStates in that gap. This then snapshots the newer baseline under the lock while installing the older (offsetTso,endTso) range, so current - stored can collapse a large cloud read and choose a broadcast build (the IVM rewrite path calls the hook outside a statement-wide stream lock). Carry a generation/offset token through installation or retry when local state advanced, and add a concurrent cloud plan/consume test.
TPC-H: Total hot run time: 28269 ms |
TPC-DS: Total hot run time: 151346 ms |
ClickBench: Total hot run time: 24.22 s |
There was a problem hiding this comment.
Static review of Apache Doris PR #69850 at exact head 6d2b9511c74f8f1d98fc04e49a96dc8c8ec7cf23 converged after two review rounds.
I independently confirmed that six existing P1 inline findings still apply at this head and carried them by ID without reposting: 4424820813 (unknown snapshot counters), 4424820822 (empty-at-create first load), 4424820835 (TTL-clipped baseline), 4424820844 (post-offset counter sampling), 4424820858 (idempotent cloud retry recording), and 4424820863 (cloud read-state/local-baseline race). The earlier P1 findings 4423124345 and 4424439892 are fixed by the current stream-lock/map normalization and known-nonempty sentinel; the existing P2 4414336957 was reviewed but not duplicated.
One distinct P1 was found: direct unestimated @incr scans use a changelog wrapper with no per-read estimate. When only the base-index row count is user-injected, StatsCalculator looks up the missing changelog-index entry, floors -1 to one row, and bypasses the wrapper's delegated changelog count, allowing a large changelog relation to become a broadcast build.
Locking, exception handling, wrapper/index delegation, stream-state persistence, cloud/local publication, TTL and unknown-range fallbacks, optimizer integration, and all changed tests were inspected. No builds or tests were executed because this review is static-only. No additional user focus was supplied. No unresolved code candidates remain after the second-round recheck.
Review converged after 2 round(s).
Existing P0/P1 findings confirmed for this head: #69850 (comment), #69850 (comment), #69850 (comment), #69850 (comment), #69850 (comment), #69850 (comment)
| */ | ||
| private double getOlapTableRowCount(OlapScan olapScan) { | ||
| OlapTable olapTable = olapScan.getTable(); | ||
| if (olapTable instanceof OlapTableWrapper) { |
There was a problem hiding this comment.
[P1] Preserve changelog row counts when base-table stats are user-injected. A direct @incr scan uses the legacy OlapChangelogTableWrapper constructors, so getRowCountEstimate() is UNKNOWN_ROW_COUNT. When a user injects only the base-index row count, StatsCalculator's userInjected branch then looks up the wrapper's changelog index and gets -1, and computeOlapScan raises that to the one-row floor without calling the wrapper's delegated changelog-index count. A large changelog can therefore be selected as a broadcast build. Fall back to the wrapper/origin changelog-index count when the injected statistic has no entry for the selected index, and add a direct @incr planner regression for this base-index-only injection.
### What problem does this PR solve? Issue Number: None Related PR: apache#69850, apache#69896 Problem Summary: Stream P/Q snapshots read partition index row counts, but changelog load deltas were stored in a separate lifetime counter in table statistics. That counter was not read by the partition lookup, lost its baseline after DROP STATS, and retained removed rows after TTL or partial TRUNCATE. Apply published changelog deltas directly to the partition index row-count estimate. Periodic tablet statistics replace the estimate, and partition replacement naturally discards it. Normalize each changelog tablet's rows using its partition replica allocation and ignore late reports for removed tablets or partitions so changelog rows cannot enter the base counter. Remove the unused lifetime counter and cover the partition lookup and statistics/partition lifecycle in AnalysisManagerTest. Cloud commits and partial publish retries remain subject to the existing asynchronous statistics limitations. A report that already includes the load may temporarily cause an overestimate until the next report. ### Release note Shared-nothing loads update partition changelog row-count estimates before periodic tablet statistics arrive. ### Check List (For Author) - Test: Unit Test - AnalysisManagerTest: 28 passed - TableStatsMetaTest: 15 passed - FE Checkstyle validation passed - Behavior changed: Yes, publish deltas update partition changelog estimates - Does this need documentation: No
…ers with
Issue Number: None
Related PR: None
Problem Summary:
A scan that reads a table stream is planned over a lightweight wrapper of the base table:
OlapTableWrapper for the base-table reads (stream@reset, and the base part of a snapshot read),
OlapChangelogTableWrapper for the changelog part. The wrapper delegates partition lookups but holds
no partitions of its own, while OlapTable derives a row count by iterating that map. Every scan over
a wrapper was therefore estimated at zero rows, which StatsCalculator.computeOlapScan raises to one:
a read of a four-billion-row table looked like a one-row scan, and the optimizer could pick the real
large side as the broadcast build, exhausting the task memory limit of an incremental view
maintenance refresh.
In cloud mode NormalizeOlapTableStreamScan wraps the base table for every read once the cloud read
states are installed, so both stream@reset() and stream@snapshot() were estimated at one row.
Fix, part one -- ask the table the wrapper reads:
- OlapTableWrapper delegates getRowCountForIndex, getRowCountForPartitionIndex, getPartitionNum and
getPartitionItemOrAnalysisException to the table it wraps. A reset read, and a snapshot read whose
offsets already reached the end, see exactly that table's content; a snapshot read of an older
offset is bounded by it. With the partition count in place, a partition-pruned scan takes the
statistics calculator's per-partition path, which looks up the selected partitions' items to
tighten the partition-key min/max.
- StatsCalculator.getOlapTableRowCount asks a wrapper which records its own estimate before it reads
the statistics record: that record describes the base index of the table, and for a changelog read
it has no entry at all, which would leave the read at the one-row floor.
Fix, part two -- record what a stream read answers with:
An incremental read answers with the change rows written after the stream's offset, and a snapshot
read with the image at that offset. Neither is the row count of the table, and the frontend cannot
derive either from the metadata: the changelog index of a partition holds the change rows of every
offset range.
- OlapTableStream records, next to each partition's offset, a PartitionStreamState: the row count of
the partition's changelog index (P) and of its base index (Q) at the moment the offset advanced
(Gson field "ps", written by initializeLocalOffsets and by unprotectedUpdateStreamUpdate; in cloud
mode TableStreamManager.recordPartitionStreamStates records it after the commit RPC succeeds).
- A changelog read is estimated with max(P_now - P_stored, P_now when the counters are unusable): a
read of an empty range is recognized by the TSOs, which do not lag behind like the row counts, and
everything else takes a value which bounds the read rather than the one-row floor.
- A snapshot read is estimated with max(Q, current base rows), or base rows plus the change rows of
the range for a unique or aggregating table, whose image also holds the rows deleted after the
offset.
Every rule errs on the large side: over-estimating only makes the optimizer plan the read
differently, while under-estimating lets a large read become the build side of a broadcast join.
The estimate relies on the tablet row counts the frontend holds, so it is a bound once the backends
have reported them (a stream records its counters from those reports, and the regression waits for
them); a load followed by a refresh before the report arrives still falls back to a value of the
right order of magnitude rather than to one row.
Fix stream reads (including incremental view maintenance refreshes) estimating every scan at one
row, which could make the optimizer broadcast a very large build side and exhaust memory during a
refresh. An incremental read of a stream is now estimated with the change rows after its offset, and
a snapshot read with the image at that offset.
- Test: Unit test / Regression test / Manual test
- Unit test: OlapTableWrapperTest (row count delegation; the changelog wrapper falls back to the
changelog row count; a recorded estimate overrides the wrapped table; partition item lookup
delegation).
- Unit test: PartitionStreamStateTest (the estimate rules: an empty range, a normal delta, no
baseline, counters moving backwards, unreported counts, an unknown range, both snapshot rules).
- Unit test: StreamScanRowCountTest (TestWithFeService: a changelog table, a stream over it and a
row count injected as ALTER TABLE ... SET STATS does; the whole planner runs and the stream scan
is estimated at the injected row count for both a snapshot and a reset read, inside a join).
- Regression test: changelog_p0/test_stream_scan_row_count, which runs in both the shared-nothing
and the cloud P0 pipelines: reset and snapshot estimates, then consuming the stream and checking
that the incremental read is planned as an empty relation when nothing was written and at three
rows after three more are loaded.
- Manual test: the reproduction below, verified through query profiles on a local cloud cluster,
and the regression suite run against a local shared-nothing cluster.
- Behavior changed: Yes. Scans over a stream read are estimated from what the read answers with
instead of one row; an incremental read is no longer unknown to the optimizer.
- Does this need documentation: No
### What problem does this PR solve? Issue Number: None Related PR: apache#69850 Problem Summary: A table stream estimates the change rows a read returns from the row count of the partition's changelog index: a read of the range between the stream offset and the end of the read returns at most the change rows the changelog still holds. The count the frontend holds comes from the tablet reports, which a backend sends every tablet_stat_update_interval_second (60s by default), so a read which follows a load within that interval sees the count from before the load. The estimate falls back to the row count of the table, which is safe but loose: it answers with the whole table where the answer is a few change rows. The frontend cannot derive the change rows from the rows it loaded into the base index: an update writes change rows without adding a base row, and a delete adds none to the base but at least one to the changelog. The backend knows them, so it now reports them. - The publish task reports the delta rows of the changelog tablets as well, not only of the base tablets: the report already visits the rowsets of every tablet a transaction wrote, and a changelog tablet is one of them. - AnalysisManager.replayUpdateRowsRecord routes those tablets to the table record by the index of each tablet (from the tablet inverted index, which the followers have as well, so a replay reproduces the routing without a new journal entry), and TableStatsMeta accumulates them per index (Gson field "iur"). The base index rows are counted as before: the change rows are neither added to them nor mistaken for them. - TableStatsMeta.getRowCountWithDeltaRows returns the accumulated rows for a changelog index: an analysis never collects its row count, so the rows of the loads are the count there. A stream read consumes this through TableStatsMeta.getRowCountWithDeltaRows, which is where the estimate of PR apache#69850 reads its counters from; until that lands, the recorded rows are visible in the table statistics only. ### Release note None ### Check List (For Author) - Test: Unit test - Unit test: TableStatsMetaTest, a new case (the rows a load writes to the changelog index are accumulated per index and reported by getRowCountWithDeltaRows, while the base index rows are not affected). TableStatsMetaTest 16/16 and AnalysisManagerTest 27/27 pass with the change. - No regression exercises the reporting window: it needs a load followed by a read within the tablet reporting interval, which the suites cannot time reliably. Every load now reports the change rows as well, so the existing suites which read the row counts of a loaded table cover the base index side of the change. - Behavior changed: No user visible change; a load additionally reports the rows it wrote to the changelog index. - Does this need documentation: No
### What problem does this PR solve? Issue Number: None Related PR: apache#69850, apache#69896 Problem Summary: Stream P/Q snapshots read partition index row counts, but changelog load deltas were stored in a separate lifetime counter in table statistics. That counter was not read by the partition lookup, lost its baseline after DROP STATS, and retained removed rows after TTL or partial TRUNCATE. Apply published changelog deltas directly to the partition index row-count estimate. Periodic tablet statistics replace the estimate, and partition replacement naturally discards it. Normalize each changelog tablet's rows using its partition replica allocation and ignore late reports for removed tablets or partitions so changelog rows cannot enter the base counter. Remove the unused lifetime counter and cover the partition lookup and statistics/partition lifecycle in AnalysisManagerTest. Cloud commits and partial publish retries remain subject to the existing asynchronous statistics limitations. A report that already includes the load may temporarily cause an overestimate until the next report. ### Release note Shared-nothing loads update partition changelog row-count estimates before periodic tablet statistics arrive. ### Check List (For Author) - Test: Unit Test - AnalysisManagerTest: 28 passed - TableStatsMetaTest: 15 passed - FE Checkstyle validation passed - Behavior changed: Yes, publish deltas update partition changelog estimates - Does this need documentation: No
### What problem does this PR solve? Issue Number: None Related PR: apache#69850 Problem Summary: Changelog index row counts can remain stale or unknown until periodic tablet statistics arrive. Base-table load rows cannot describe changelog writes: updates and deletes have their own physical changelog records. Cloud commit responses previously carried no partition/index-level changelog delta, and default cloud replica zeros could be marked as fully reported counts. This PR completes physical changelog row-count collection in shared-nothing and cloud deployments: - Read attached changelog rowsets after publish workers finish, including segments appended by merge-on-write publication. Normalize shared-nothing deltas by each partition's replica allocation and filter removed or replaced tablets. - Return optional partition/index-level changelog deltas from ordinary, lazy and subtransaction cloud commits, with tablet identities and the final partition version. Only completed visible commits supply additive statistics. Cloud rows are counted once without replica division and remain separate from base updatedRows. - Preserve unknown counts on zero deltas and establish positive partial estimates without setting the complete-report flag. Apply duplicate and out-of-order cloud responses once using compact version ranges; complete periodic reports retire superseded versions. - Overwrite retained counts from valid periodic tablet statistics so TTL can lower the count. Preserve partial-report validity and serialize count publication with commit updates under the table metadata lock. Replacing tablet data resets its local estimate; DROP STATS preserves it. Counts are physical records; UPDATE records are not multiplied by two. Periodic reports and commit deltas can temporarily overlap or lag. Failed publish attempts can miss immediate deltas, which later tablet reports calibrate; these estimates are not exact historical counts or strict upper bounds. ### Release note Successful loads update partition changelog row-count estimates in both shared-nothing and cloud deployments before the next periodic tablet report. ### Check List (For Author) - Test: Unit Test / Regression test - 121 focused FE tests covering analysis statistics, index counts, periodic reports, cloud commit callbacks and retries passed across focused runs. - ASAN BE publish test passed with real concurrent MoW writes, publish-time changelog segment growth, final row-count reporting and publish retry. - 5 ASAN meta-service RPC tests passed for ordinary, lazy and subtransaction commits, failures/retries, final temporary rowset metadata and base/changelog separation. - Shared-nothing changelog_p0 regression directory: 27 suites passed. The new suite pauses periodic statistics publication and checks immediate counts, stream results, DROP STATS and TRUNCATE. - FE build (excluding unavailable OBS dependencies), ASAN BE build, Checkstyle, clang-format, clang-tidy, build hygiene and cross-release protobuf checks passed. - Cloud end-to-end cluster regression was not run; cloud coverage consists of the RPC and FE tests above. - Behavior changed: Yes, successful loads immediately supplement partition changelog row-count estimates in both storage modes, while periodic statistics continue to calibrate retained stock. - Does this need documentation: No
### What problem does this PR solve?
Issue Number: None
Problem Summary: Stream wrappers can lose row-count statistics, while whole-table counts do not describe incremental ranges or historical snapshots. Estimate selected partitions from the larger valid base/changelog difference, apply actual scan-type expansion once, and use the consumption base count for DUP snapshots. Bind local baselines to offsets and data lifecycles, invalidate recovery state, preserve unknown and partial statistics, and prevent injected base statistics from overriding explicit scan estimates.
### Release note
Improve stream row-count estimates for incremental and historical snapshot reads, including cloud read-state matching.
### Check List (For Author)
- Test: Unit test / Regression test
- 105 related FE tests passed.
- 17 stream regression suites passed; row-count golden output generated by the runner.
- FE Checkstyle/build and ASAN BE build passed.
- Behavior changed: Yes. Estimates follow the selected read and consumption baseline.
- Does this need documentation: No
6d2b951 to
4b09d50
Compare
yujun777
changed the title
### Background Stream row-count baselines are optional FE-local statistics recorded during creation and successful consumption. Source metadata can change independently of the sink and stream locks held by the consuming transaction. ### What problem does this PR solve? Issue Number: None Related PR: None PR Type: Bugfix Problem Summary: Optional baseline sampling could race source DDL after durable local commit or deadlock cloud publication with a metadata reader and a queued source writer. Protect source metadata with nonblocking locks and fill the missing planner and cloud integration tests. ### Details - **Change map:** 48 production lines and 318 test lines; 0 generated or documentation lines. Source sampling/publication lock handling accounts for 45 production lines; removing an unreachable statistics branch accounts for 3. - **Frequency:** Not observed in a deployed cluster; deterministic unit tests reproduce unlocked source sampling, publication during source DDL and the three-thread fair-lock cycle. - **Root cause:** Local consumption sampled partitions without a source lock. Cloud publication held source READ while waiting indefinitely for stream WRITE, conflicting with consumption metadata readers and source DDL. - **Why existing tests missed it:** Local snapshot/update tests did not schedule concurrent source DDL. Cloud commit tests used unregistered streams and empty samples. Existing regression suites checked stream results without concurrent metadata-lock schedules or unknown/mixed calculator inputs. - **Fix:** Try source READ without waiting for local/cloud sampling and cloud publication; try stream WRITE without waiting under source READ. Skip unavailable auxiliary samples and retain ordinary fallback. Add deterministic concurrency tests, calculator unknown/mixed cases, production cloud success/failure/replacement/duplicate/late cases and authoritative CREATE-offset binding. - **Other paths and branches:** Shared-nothing and cloud consumption are fixed. Creation retains its existing source lock protocol. Checked branch-3.1, branch-4.0, branch-4.1 and branch-4.2: the stream/wrapper classes are absent. ### Testing - 115 focused FE tests passed locally across nine classes. The three concurrency tests fail on the preceding PR revision and pass with the fix. - Final FE build and Checkstyle passed locally; all FE test sources compiled after the local-variable cleanup. - All 17 stream regression suites passed locally on the rebuilt FE, including row estimates and ordered query results. ### Release note None ### Check List (For Author) - Behavior changed: No - Does this need documentation: No - Pre-submission status: Ready ### AI Disclosure - Tool: Codex; gpt-6.1-sol/xhigh for independent reviews - Generated: Code, tests, and this description - Human review: not stated
### Background Visible loads supplement changelog index counts before periodic tablet statistics. Stream consumption advances its own offset independently of those current counts. ### What problem does this PR solve? Issue Number: None Related PR: apache#69850 (stream consumption baselines) PR Type: Test Problem Summary: Cover successive loads and stream consumptions while periodic row-count publication is paused. Current physical counts must grow after each load and remain unchanged by consumption; each successful consumption must empty its interval. ### Details - **Scenarios:** Extend the existing SN/cloud immediate-count suite with two load/consume cycles. Generated golden sections show retained counts 2, 3 and 4, the two new rows in the sink, and empty intervals after consumption. - **What existing tests miss:** The existing paused-report test performed only SELECT from the stream, without advancing its offset between writes. - **Fails when broken:** A missing immediate delta changes the recorded index count; failed offset advancement leaves a nonempty interval or repeats rows in the sink. ### Testing The extended suite passed on local shared-nothing and ASAN cloud clusters against the same runner-generated golden file. No production code changed in this commit. Four previously reported cloud row-content failures elsewhere remain outside this verification. ### Release note None ### Check List (For Author) - Behavior changed: No - Does this need documentation: No - Pre-submission status: Ready ### AI Disclosure - Tool: Codex - Generated: Regression additions, generated golden output, and this description - Human review: not stated
### Background Table streams read changelog ranges or reconstruct snapshots at a consumption offset. Optimizer statistics originate from base and changelog indexes and can lag data visibility. Each stream needs its own consumption-bound baseline independently of current index statistics. ### What problem does this PR solve? Issue Number: None Related PR: apache#62058 (introduced the OLAP wrapper), apache#69896 (supplements current changelog counts after visible loads) PR Type: Bugfix Problem Summary: Stream wrappers can size large reads as one row or use retained changelog stock as an incremental range. Estimate each read from its selected partitions, actual scan type and consumption-bound counts, with lifecycle checks and ordinary statistics fallback. Store the consumed offset, base count and changelog count in one partition state, prepare that state before transaction persistence, and publish the same tuple only after success. Shared-nothing transactions and cloud MS offsets persist the tuple through their existing storage paths. ### Details - **Change map:** production +790/-40, tests +1266/-23, generated golden output +44/-0, no repository documentation changes. Production changes cover stateless estimate formulas and wrapper/statistics plumbing; one persisted partition consumption state and source lifecycle binding; existing local transaction and cloud creation/commit integration; and nine optional protobuf fields with atomic MS offset storage/readback. Tests cover estimates, lifecycle/concurrency, transaction replay/abort and cloud/MS storage boundaries. - **Frequency:** Affected wrapper reads can collapse to one row; range quality depends on available statistics. Nonempty offsets with stale, zero or negative deltas require a stock fallback instead of proving zero rows. - **Root cause:** Wrappers inherited empty partition maps, while retained index counts do not describe a consumed range. Separate or post-commit baseline sampling can lose the association between the offset and its sampled counts. - **Why existing tests missed it:** Existing tests checked metadata delegation, SQL results and offset advancement without planner count assertions or durable offset/count tuples. Cloud transaction fixtures used unregistered streams and did not cover nonempty production capture/publication, report changes during RPCs, or the fair-lock cycle reproduced during review. - **Fix:** Persist offset/B0/C0 together in PartitionStreamState and pending OlapTableStreamUpdate. Sample before persistence/RPC; publish unchanged samples after success, leave failed/aborted updates unpublished, and retain counts for unchanged consumed offsets. Use positive changelog/base deltas, then stock/table fallback; only offsets prove an empty range. Apply actual scan-type expansion once and preserve reset/snapshot branch scales. Source sampling and cloud publication use nonblocking metadata locks; publication takes no source lock. FE/data lifecycle bindings invalidate stale estimates without discarding persisted numbers. - **Other paths and branches:** Covers SN/cloud creation, consumption, image/replay, MS latest/versioned offset storage and existing read states. First initial-snapshot consumption can transition to consumed at the same numeric offset. Column statistics and BE/MS current-count collection remain separate; apache#69896 updates current C1 without changing stream baselines. Stream/IVM storage is unreleased, so no old-format JSON migration is added. All new protocol fields are optional with unused IDs; stream messages are absent from branch-3.1/4.0/4.1/4.2. - **Behavior change:** Stream scan estimates use the consumed range and selected partitions. Offset and sampled counts are durable together; unknown or invalid baselines use existing fallback rules. Counts remain approximate, and completely unknown inputs retain ordinary OLAP handling. ### Testing - Local FE build and Checkstyle passed. 144 distinct focused FE tests passed across eleven classes, including actual TransactionState serialization/replay and abort, cloud registered-source commit/create hooks, unchanged offsets, first full consumption at an unchanged numeric TSO, and a nonempty range between reports. - ASAN Cloud build passed. All 42 MetaServiceTableStreamTest cases passed, covering offset/count storage and readback, restart, failed/CAS-conflicting commits, idempotence, unchanged consumed offsets and latest/versioned storage. Clang-format v16 and changed-line clang-tidy passed. - All 17 local SN stream regression suites passed, including planner estimates and ordered query results. The three prior concurrency reproductions failed before their fix and passed afterwards. The ASAN BE build passed earlier; no BE execution source changes are included. - Cloud FE paths used mocked MS RPCs and real in-memory MS transaction tests; this PR's new baseline protocol was not exercised on a live cloud cluster. The companion apache#69896 paused-report regression passed in SN and cloud with two successive load/consume cycles, against runner-generated output. ### Notes for Reviewers C0 is retained physical changelog stock, not the number of consumed rows. Initial full reads use base scan estimates. Replay preserves stored counts but clears their transient FE/source binding until a later successful consumption establishes one. Periodic/current-count updates never write stream baselines; report/delta overlap and lag remain approximation limits. ### Release note Table stream scans retain useful row estimates for incremental and snapshot reads, and persist sampled consumption counts together with their offsets. ### Check List (For Author) - Behavior changed: Yes (see Details) - Does this need documentation: No - Pre-submission status: Ready ### AI Disclosure - Tool: Codex; gpt-6.1-sol/xhigh for independent reviews - Generated: Code, tests, and this description - Human review: not stated
Local pipeline review — ✅ PASSschema: doris-repo-review/v1
status: PASS
pr: apache/doris#69850
commit: b78642cd9bed2ce8b3a98607fe9df9bed9a3e2b7
base: ee73acc6131e6c6c752eb59e252404590be6ad77
reviewed_at: 2026-11-13T20:11+08:00
reviewer: yujun777
model: gpt-6.1-sol
effort: xhigh
findings: {blocker: 0, major: 0, minor: 0, nit: 0}
rounds: 1
converged: trueNotes for maintainers
Reviewed locally with the |
### Background Stream updates and scan wrappers keep partition offsets together with consumption counts or read metadata. Their parallel maps share partition IDs, but key sets differ for first consumption, empty partitions, cloud UNKNOWN states and optional scan statistics. ### What problem does this PR solve? Issue Number: None Related PR: apache#69896 (current changelog row counts) PR Type: Refactor Problem Summary: Store each partition's related update or scan fields in one map instead of separate maps. Preserve absent previous offsets, ranges, baselines, fixed versions and estimates, including their distinct projected key sets. ### Details - **Change map:** production +204/-183 and tests +118/-17. No protocol, generated output or repository documentation changes. Three state containers and their consumers are grouped by partition; tests check distinct field memberships, persistence, merge and identity. - **Moves vs. logic changes:** Replace two update maps with a map of previous offset and next complete state. Replace three maps in each wrapper with one map of partition metadata. Adapt consumption and publication to the grouped representation; keep existing merge, CAS, estimate and identity behavior. - **Behavior change:** No SQL or estimation behavior changes. Explicit row counts remain excluded from wrapper equality/hash; offset-only and version-only partitions are excluded from estimated totals. - **Compatibility:** The unreleased stream update JSON uses a single updates field. No migration for intermediate unreleased formats. FE/MS and FE/BE protocols are unchanged. - **Performance impact:** One stored map per class. Preserve direct O(1) partition lookups and cached presence flags. Existing map interfaces use lazy views rather than copying complete maps on every call. No benchmark claim. - **Rollback:** Revert this refactor before release; intermediate unreleased JSON is not migrated. ### Testing 199 focused FE tests across thirteen classes passed, including new mixed previous-offset merge/serialization/CAS tests, empty/UNKNOWN read memberships, disjoint scan metadata fields and estimate-free identity. Local FE build and Checkstyle passed. All 17 existing SN stream regression suites passed against unchanged golden files. No native source or protocol changes in this commit. ### Release note None ### Check List (For Author) - Behavior changed: No - Does this need documentation: No - Pre-submission status: Ready ### AI Disclosure - Tool: Codex; gpt-6.1-sol/xhigh for independent reviews - Generated: Refactor, tests and this description - Human review: not stated
### Background Stream updates and scan wrappers carry per-partition offsets, row-count baselines or read metadata. Their optional fields have different memberships, so one partition state must preserve which fields exist. ### What problem does this PR solve? Issue Number: None Related PR: apache#69896 (current changelog counts) PR Type: Refactor Problem Summary: Use named update values and one state map at construction boundaries. Remove split-map constructor overloads, use compact persisted names, and express initial/restored consumption states through explicit factories. Preserve read memberships, offset CAS and estimate-free wrapper identity. ### Details - **Change map:** Production +143/-120, tests +174/-128; no protocol or generated output changes. Three state APIs and their construction/consumption callers; tests adapt fixtures, check compact JSON and request an early BE tablet report before polling FE counts. - **Moves vs. logic changes:** Replace map-value Pair with immutable PartitionUpdate(previousOffset,nextState), and serialize it under u/p/n. Wrapper/update constructors accept one named-state map. PartitionStreamState has one complete internal constructor, an immutable initial seed and an explicit restored-fields factory; existing transitions preserve epoch/count semantics. - **Behavior change:** SQL behavior and read estimates are unchanged. The unreleased update JSON uses shorter names. The regression waiter requests an SN BE tablet report before its existing FE statistics condition. - **Compatibility:** No migration for intermediate unreleased stream JSON, as the feature is not externally released. All Java construction callers are adapted; FE/BE and FE/MS protocols are unchanged. - **Performance impact:** Retain expected-size allocation in extraction, update construction/commit copying and scan-state assembly. Immutable initial states share an empty seed; map views avoid full copies for existing read interfaces. No benchmark claim. - **Rollback:** Revert this refactor before release; intermediate unreleased JSON is not migrated. ### Testing Local FE build, Checkstyle and all test-source compilation passed. 199 focused FE tests across thirteen classes and 17 existing SN stream regression suites passed against unchanged golden files. Tests cover JSON short fields plus type discriminator, merge/roundtrip/CAS, initial and replay states, wrapper membership/identity, and planner/transaction/cloud creation and commit callers. ### Release note None ### Check List (For Author) - Behavior changed: Yes (see Details) - Does this need documentation: No - Pre-submission status: Ready ### AI Disclosure - Tool: Codex; gpt-6.1-sol/xhigh for independent reviews - Generated: Refactor, tests and this description - Human review: not stated
Local pipeline review — ✅ PASSschema: doris-repo-review/v1
status: PASS
pr: apache/doris#69850
commit: 54c5ef1bb7a2fd8209d9edea633020179a550b0a
base: ee73acc6131e6c6c752eb59e252404590be6ad77
reviewed_at: 2026-11-13T23:36+08:00
reviewer: yujun777
model: gpt-6.1-sol
effort: xhigh
findings: {blocker: 0, major: 0, minor: 0, nit: 0}
rounds: 1
converged: trueNotes for maintainers
Reviewed locally with the |
|
run buildall |
Background
Table streams read changelog ranges or reconstruct snapshots at a consumption offset. Optimizer statistics originate from base and changelog indexes and can lag data visibility. Each stream needs its own consumption-bound baseline independently of current index statistics.
What problem does this PR solve?
Issue Number: None
Related PR: #62058 (introduced the OLAP wrapper), #69896 (supplements current changelog counts after visible loads)
PR Type: Bugfix
Problem Summary:
Stream wrappers can size large reads as one row or use retained changelog stock as an incremental range. Estimate each read from its selected partitions, actual scan type and consumption-bound counts, with lifecycle checks and ordinary statistics fallback. Store the consumed offset, base count and changelog count in one partition state, prepare that state before transaction persistence, and publish the same tuple only after success. Shared-nothing transactions and cloud MS offsets persist the tuple through their existing storage paths. Updates and wrappers each accept one partition map of named state objects, with optional fields for different read memberships.
Details
Testing
Notes for Reviewers
C0 is retained physical changelog stock, not the number of consumed rows. Initial full reads use base scan estimates. Replay preserves stored counts but clears their transient FE/source binding until a later successful consumption establishes one. Periodic/current-count updates never write stream baselines; report/delta overlap and lag remain approximation limits.
Release note
Table stream scans retain useful row estimates for incremental and snapshot reads, and persist sampled consumption counts together with their offsets.
Check List (For Author)
AI Disclosure